package InvertedIndex

import org.apache.spark.{SparkContext, SparkConf}
import scala.collection.mutable


object   Invertedindex {
    def main (args: Array[String]) {
     val conf = new SparkConf().setAppName("Inverted Index")
      //本地单线程运行
      if (!conf.contains("spark.master"))
        conf.setMaster("local")
      val sc =new SparkContext(conf)
      //SparkContext.wholeTextFiles允许你读取文件夹下所有的文件，比如多个小的文本文件， 返回文件名/内容对。
      val files =sc.wholeTextFiles("input")
      //因为读进来的文件是路径，所以要把文件名过滤一下，前面的都去掉，只留下文件自己的名字
      val file_name_length = files.map(x=>x._1.split("/").length).collect()(0)
      val file_name_context= files.map(x=>(x._1.split("/")(file_name_length-1),x._2)).sortByKey()
      //words为我们最后得到的（文件名，单词）对
      val words = file_name_context.flatMap(
        x=>{
          val line = x._2.split("\n")
          val list = mutable.LinkedList[((String,String),Int)]()
          var temp = list
          for (i<- 0 until line.length){
            val word = line(i).split(" ").iterator
            while (word.hasNext){
              temp.next = mutable.LinkedList[((String,String),Int)](((x._1,word.next()),1))
              temp = temp.next
            }
          }
          val list_end=list.drop(1)
          list_end
        }
      )

      //首先进行wordcount,之后按照文件名排序，然后调换map的位置，将文件名串起来，再根据单词排序，最后保存
      words.reduceByKey(_+_).sortByKey().map(x=>(x._1._2,(x._1._1,x._2))).groupByKey().map({
        case (x, y)=> "\""+x+"\" : {"+ y.mkString(",")+"}"
      }).saveAsTextFile("output")
      sc.stop()
   }

 }
